package com.gmall.app

import com.gmall.bean.OrderInfo
import com.gmall.utils.{Constants, PropertiesUtil}
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream
import org.json4s.JsonAST.JString
import org.json4s.{CustomSerializer, DefaultFormats, Formats, JDouble, JValue}
import org.json4s.jackson.JsonMethods

/**
 * <p>项目名称: 实时数仓项目 </p>
 * <p>文件名称: 统计每日交易额 </p>
 * <p>描述: 销售额总数、销售额分时统计 </p>
 * <p>创建时间: 2020/10/19</p>
 * @author <a href="mail to: 1034652894@qq.com" rel="nofollow">Demik Qin</a>
 * @version v1.0
 * @update [序号][日期YYYY-MM-DD] [更改人姓名][变更描述]
 */
object OrderApp extends BaseApp {
    override val appName: String = "OrderApp"
    override val master: String = "local[*]"
    override val batchTime: Int = 3
    override val groupId: String = "OrderApp"
    override val topic: String = Constants.ORDER_INFO_TOPIC

    private val zkUrl: String = PropertiesUtil.getValue("zkUrl")

    /**
     * 自定义String => Double解析器
     */
    val StringToDouble: CustomSerializer[Double] = new CustomSerializer[Double](format => ( {
        case JString(s) => s.toDouble
    }, {
        case d: Double => JDouble(d)
    }
    ))

    /**
     * 业务操作
     * @param ssc
     * @param sourceStream kafka数据源
     */
    override def run(ssc: StreamingContext, sourceStream: DStream[String]): Unit = {
        val orderInfoStream = sourceStream.map(json => {
            //println(json)
            // 解析json字符串
            implicit val f: Formats = org.json4s.DefaultFormats + StringToDouble
            JsonMethods.parse(json).extract[OrderInfo]
        })
        //orderInfoStream.print(100)
        orderInfoStream.foreachRDD(rdd => {
            // 将数据保存到phoenix
            import org.apache.phoenix.spark._
            rdd.saveToPhoenix(
                // 表名
                tableName = "GMALL_ORDER_INFO",
                // 列名，与样例类字段的顺序保持一致
                Seq("ID", "PROVINCE_ID", "CONSIGNEE", "ORDER_COMMENT", "CONSIGNEE_TEL", "ORDER_STATUS", "PAYMENT_WAY", "USER_ID", "IMG_URL", "TOTAL_AMOUNT", "EXPIRE_TIME", "DELIVERY_ADDRESS", "CREATE_TIME", "OPERATE_TIME", "TRACKING_NO", "PARENT_ORDER_ID", "OUT_TRADE_NO", "TRADE_BODY", "CREATE_DATE", "CREATE_HOUR"),
                zkUrl = Option(zkUrl)
            )
        })
    }


}